package top.jolyoulu.streaming.example

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext, StreamingContextState}

import java.util.Properties
import scala.collection.mutable.ListBuffer
import scala.util.Random

/**
 * @Author: JolyouLu
 * @Date: 2024/5/19 15:07
 * @Description
 */
object Spark01_SparkStreaming_MockData {
  def main(args: Array[String]): Unit = {
    //生成模拟数据
    //格式：timestamp area city userid adid
    //说明：时间错 区域 城市 用户 广告
    //生成数据推送到Kafka，SparkStreaming分析

    val kafkaProducer: KafkaProducer[String, String] = createKafkaProducer("192.168.88.100:9092")

    while (true){
      //生成数据
      mockData.foreach(
        data => {
          //向kafka发送数据
          kafkaProducer.send(new ProducerRecord[String, String]("mockData",data))
//          println(data)
        }
      )
      Thread.sleep(2000)
    }
  }

  //生成数据
  def mockData():ListBuffer[String] = {
    val list = ListBuffer[String]()
    val areaList = ListBuffer[String]("华东","华南","华北")
    val cityList = ListBuffer[String]("北京","上海","广州","深圳")
    for (i <- 1 to 30){
      val area = areaList(new Random().nextInt(3))
      val city = cityList(new Random().nextInt(4))
      val userId = new Random().nextInt(6) + 1
      val acId = new Random().nextInt(6) + 1

      list.append(s"${System.currentTimeMillis()} ${area} ${city} ${userId} ${acId}")
    }
    list
  }

  //创建一个Kafka生产者
  def createKafkaProducer(broker:String): KafkaProducer[String,String] = {
    // 创建配置对象
    val prop = new Properties()
    // 添加配置
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    new KafkaProducer[String,String](prop)
  }
}
